 1.Kafka高级特性解析之主题中的必要参数配置
   
   kafka-topics.sh --config xx=xx --config yy=yy
   配置给主题的参数。
   属性                       默认值      服务器默认属性     
   cleanup.policy             delete      log.cleanup.policy
   说明：要么是”delete“要么是”compact“； 这个字符串指明了针对
旧日志部分的利用方式；默认方式（"delete"）将会丢弃旧的部分当
他们的回收时间或者尺寸限制到达时。”compact“将会进行日志压缩
   compression.type            none
   说明：producer用于压缩数据的压缩类型。默认是无压缩。正确的选项值是none、gzip、
snappy、lz4。压缩最好用于批量处理，批量处理消息越多，压缩性能越好。
   max.message.bytes         1000000      max.message.bytes
   说明：kafka追加消息的最大字节数。注意如果你增大这个字节数，也必须增大consumer的
fetch字节数，这样consumer才能fetch到这些最⼤字节数的消息。
   min.cleanable.dirty.ratio   0.5        min.cleanable.dirty.ratio
   说明：此项配置控制log压缩器试图进行清除日志的频率。默认情况
下，将避免清除压缩率超过50%的日志。这个比率避免了最大的空间浪费
   min.insync.replicas          1         min.insync.replicas
   说明：当producer设置request.required.acks为-1时,min.insync
.replicas指定replicas的最小数目(必须确认每一个repica的写数
据都是成功的),如果这个数目没有达到，producer会产生异常。
   retention.bytes             None       log.retention.bytes
   说明：如果使用“delete”的retention 策略，这项配置就是指在删除日志
之前,日志所能达到的最大尺寸。默认情况下，没有尺寸限制而只有时间限制
   retention.ms                 7         days log.retention.minutes
   说明：如果使用“delete”的retention策略，这项配置就是指删除日志前日志保存的时间。
   segment.bytes               1GB        log.segment.bytes
   说明：kafka中log日志是分成一块块存储的，此配置是指log日志划分成块的大小
   segment.index.bytes        10MB        log.index.size.max.bytes
   说明：此配置是有关offsets和文件位置之间映
射的索引文件的大小；置一般不需要修改这个配
   segment.jitter.ms            0         log.roll.jitter.{ms,hours}
   说明：The maximum jitter to subtract from logRollTimeMillis.
   segment.ms                7 days       log.roll.hours
   说明：即使log的分块文件没有达到需要删除、压缩的大小，
一旦log 的时间达到这个上限,就会强制新建一个log分块文件
   unclean.leader.election    true
   .enable 
   说明：指明了是否能够使不在ISR中replicas设置用来作为leader
 
 2.KafkaAdminClient应用
   
   说明
   除了使⽤Kafka的bin目录下的脚本工具来管理Kafka，还可以使用管理Kafka的API将某些管理查看的功能集成到系
统中。在Kafka0.11.0.0 版本之前,可以通过kafka-core包（Kafka的服务端，采用Scala编写）中的AdminClient和
AdminUtils来实现部分的集群管理操作.Kafka0.11.0.0 之后,又多了一个AdminClient,在kafka-client包下，一个抽
象类，具体的实现是org.apache.kafka.clients.admin.KafkaAdminClient。
   功能与原理介绍
   Kafka官网：The AdminClient API supports managing and inspecting topics, brokers, acls, and other 
Kafka objects。
   KafkaAdminClient包含了以下几种功能(以Kafka1.0.2 版本为准)：
   1).创建主题：
   createTopics(final Collection<NewTopic> newTopics, final CreateTopicsOptions options)
   2).删除主题：
   deleteTopics(final Collection<String> topicNames, DeleteTopicsOptions options)
   3).列出所有主题：
   listTopics(final ListTopicsOptions options)
   4).查询主题：
   describeTopics(final Collection<String> topicNames, DescribeTopicsOptions options)
   5).查询集群信息：
   describeCluster(DescribeClusterOptions options)
   6).查询配置信息：
   describeConfigs(Collection<ConfigResource> configResources, final DescribeConfigsOptions
options)
   7).修改配置信息：
   alterConfigs(Map<ConfigResource, Config> configs, final AlterConfigsOptions options)
   8).修改副本的日志目录：
   alterReplicaLogDirs(Map<TopicPartitionReplica, String> replicaAssignment, final
AlterReplicaLogDirsOptions options)
   9).查询节点的日志目录信息：
   describeLogDirs(Collection<Integer> brokers, DescribeLogDirsOptions options)
   10).查询副本的日志目录信息：
   describeReplicaLogDirs(Collection<TopicPartitionReplica> replicas,
DescribeReplicaLogDirsOptions options)
   11).增加分区：
   createPartitions(Map<String, NewPartitions> newPartitions, final CreatePartitionsOptions
options)
   用到的参数：
   属性                       重要性  说明
   bootstrap.servers           high   向Kafka集群建立初始连接用到的host/port列表。
                                      客户端会使用这里列出的所有服务器进行集群其他服
                                      务器的发现，而不管是否指定了哪个服务器作作引导。
                                      这个列表仅影响用来发现集群所有服务器的初始主机。
                                      字符串形式：host1:port1,host2:port2,...
                                      由于这组服务器仅用于建立初始链接，然后发现集群中的
                                      所有服务器,因此没有必要将集群中的所有地址写在这里。
                                      一般最好两台，以防其中一台宕掉。
   client.id                  medium  生产者发送请求的时候传递给broker的id字符串。
                                      用于在broker的请求日志中追踪什么应用发送了什么消息。
                                      一般该id是跟业务有关的字符串。
   connections.max.idle.ms    medium  当连接空闲时间达到这个值,就关闭连接.long型数据，默认；300000
   receive.buffer.bytes       medium  TCP接收缓存（SO_RCVBUF），如果设置为-1,则使用操作系
                                      统默认的值。int类型值，默认65536，可选值：[-1,...]
   request.timeout.ms         medium  客户端等待服务端响应的最大时间。如果该时间超时，则客户端要么
                                      重新发起请求,要么如果重试耗尽,请求失败.int类型值,默认：120000
   security.protocol          medium  跟broker通信的协议：PLAINTEXT, SSL, SASL_PLAINTEXT, 
                                      SASL_SSL.string类型值，默认：PLAINTEXT
   send.buffer.bytes          medium  用于TCP发送数据时使用的缓冲大小(SO_SNDBUF)，
                                      -1 表示使用OS默认的缓冲区大小。
                                      int类型值，默认值：131072
   reconnect.backoff.max.ms    low    对于每个连续的连接失败，每台主机的退避将成倍增加,直至达到此
                                      最大值.在计算退避增量之后,添加20％的随机抖动以避免连接风暴。
                                      long型值，默认1000，可选值：[0,...]
   reconnect.backoff.ms        low    重新连接主机的等待时间。避免了重连的密集循环。
                                      该等待时间应用于该客户端到broker的所有连接。
									  long型值，默认：50
   retries                     low    The maximum number of times to retry a call
                                      before failing it.重试的次数,达到此值,失败。
									  int类型值，默认5。
   retry.backoff.ms            low    在发生失败的时候如果需要重试,则该配
                                      置表示客户端等待多长时间再发起重试
									  该时间的存在避免了密集循环。
                                      long型值，默认值：100。
   主要操作步骤：
   客户端根据方法的调用创建相应的协议请求,比如创建Topic的createTopics方法，其内部就是发送
CreateTopicRequest请求。
   客户端发送请求至Kafka Broker。
   Kafka Broker处理相应的请求并回执,比如与CreateTopicRequest对应的是CreateTopicResponse。
客户端接收相应的回执并进行解析处理。
   和协议有关的请求和回执的类基本都在org.apache.kafka.common.requests包中，AbstractRequest和
AbstractResponse是这些请求和响应类的两个父类。
   综上，如果要自定义实现一个功能，只需要三个步骤：
   1).自定义XXXOptions;
   2).自定义XXXResult返回值；
   3).自定义Call，然后挑选合适的XXXRequest和XXXResponse来实现Call类中的3个抽象方法。
   com.lagou.kafka.demo.MyAdminClient
package com.lagou.kafka.demo;

import org.apache.kafka.clients.admin.*;
import org.apache.kafka.common.KafkaFuture;
import org.apache.kafka.common.Node;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.TopicPartitionInfo;
import org.apache.kafka.common.config.ConfigResource;
import org.apache.kafka.common.requests.DescribeLogDirsResponse;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.util.*;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.function.BiConsumer;
import java.util.function.Consumer;

public class MyAdminClient {

    private KafkaAdminClient client;

    @Before
    public void before() {
        Map<String, Object> configs = new HashMap<String, Object>();
        configs.put("bootstrap.servers", "linux121:9092");
        configs.put("client.id", "admin_001");
        client = (KafkaAdminClient) KafkaAdminClient.create(configs);

    }

    @After
    public void after() {
        // 关闭客户端
        client.close();
    }

    @Test
    public void testListTopics() throws ExecutionException, InterruptedException {
        // 列出主题
//        final ListTopicsResult listTopicsResult = client.listTopics();
        ListTopicsOptions options = new ListTopicsOptions();
        // 列出内部主题
        options.listInternal(true);
        // 设置请求超出时间，单位毫秒
        options.timeoutMs(500);
        ListTopicsResult listTopicsResult = client.listTopics(options);
//        final Set<String> strings = listTopicsResult.names().get();
//        strings.forEach(name -> {
//            System.out.println(name);
//        });
        // 将请求变成同步请求，直接获取结果
        final Collection<TopicListing> topicListings = listTopicsResult.listings().get();
        topicListings.forEach(new Consumer<TopicListing>() {
            @Override
            public void accept(TopicListing topicListing) {
                // 该主题是否是内部主题
                final boolean internal = topicListing.isInternal();
                // 该主题的名字
                final String name = topicListing.name();
                System.out.println("主题是否是内部主题:" + internal);
                System.out.println("主题的名字：" + name);
                System.out.println(topicListing);
                System.out.println("======================================");
            }
        });
    }
    // 查询节点的⽇志⽬录信息
    @Test
    public void testDescribeLogDirs() throws ExecutionException, InterruptedException {
        final DescribeLogDirsResult describeLogDirsResult = client.describeLogDirs(Collections.singleton(0));
        final Map<Integer, Map<String, DescribeLogDirsResponse.LogDirInfo>> integerMapMap = describeLogDirsResult.all().get();
        integerMapMap.forEach(new BiConsumer<Integer, Map<String, DescribeLogDirsResponse.LogDirInfo>>() {
            @Override
            public void accept(Integer integer, Map<String, DescribeLogDirsResponse.LogDirInfo> stringLogDirInfoMap) {
                System.out.println("broker.id = " + integer);
                //log.dirs可以设置多个目录
                stringLogDirInfoMap.forEach(new BiConsumer<String, DescribeLogDirsResponse.LogDirInfo>() {
                    @Override
                    public void accept(String s, DescribeLogDirsResponse.LogDirInfo logDirInfo) {
                        System.out.println("logdirs = " + s);
                        final Map<TopicPartition, DescribeLogDirsResponse.ReplicaInfo> replicaInfos = logDirInfo.replicaInfos;
                        replicaInfos.forEach(new BiConsumer<TopicPartition, DescribeLogDirsResponse.ReplicaInfo>() {
                            @Override
                            public void accept(TopicPartition topicPartition, DescribeLogDirsResponse.ReplicaInfo replicaInfo) {
                                System.out.println("主题分区：" + topicPartition.partition());
                                System.out.println("主题:" + topicPartition.topic());
                                final boolean isFuture = replicaInfo.isFuture;
                                final long offsetLag = replicaInfo.offsetLag;
                                final long size = replicaInfo.size;
                            }
                        });
                    }
                });
            }
        });
    }

    // 创建主题
    @Test
    public void testCreateTopic() throws ExecutionException, InterruptedException {
        Map<String, String> configs = new HashMap<>();
        configs.put("max.message.bytes", "1048576");
        configs.put("segment.bytes", "1048576000");
        NewTopic newTopic = new NewTopic("admin_tp_01", 2, (short) 1);
        newTopic.configs(configs);
        CreateTopicsResult topics = client.createTopics(Collections.singleton(newTopic));
//        KafkaFuture<Void> all = topics.all();
        Void aVoid = topics.all().get();
        System.out.println(aVoid);
    }

    // 删除主题
    @Test
    public void testDeleteTopic() throws ExecutionException, InterruptedException {
        DeleteTopicsOptions options = new DeleteTopicsOptions();
        options.timeoutMs(500);
        DeleteTopicsResult deleteResult = client.deleteTopics(Collections.singleton("admin_tp_01"), options);
        deleteResult.all().get();
    }

    //修改主题
    @Test
    public void testAlterTopic() throws ExecutionException, InterruptedException {
        NewPartitions newPartitions = NewPartitions.increaseTo(5);
        Map<String, NewPartitions> newPartitionsMap = new HashMap<>();
        newPartitionsMap.put("admin_tp_01", newPartitions);
        CreatePartitionsOptions option = new CreatePartitionsOptions();
        // Set to true if the request should be validated without creating new partitions.
        // 如果只是验证，而不创建分区，则设置为true
        // option.validateOnly(true);
        CreatePartitionsResult partitionsResult = client.createPartitions(newPartitionsMap, option);
        Void aVoid = partitionsResult.all().get();
        System.out.println(aVoid);
    }

    // 查询主题
    @Test
    public void testDescribeTopics() throws ExecutionException, InterruptedException {
        DescribeTopicsOptions options = new DescribeTopicsOptions();
        options.timeoutMs(3000);
        DescribeTopicsResult topicsResult = client.describeTopics(Collections.singleton("admin_tp_01"), options);
        Map<String, TopicDescription> stringTopicDescriptionMap = topicsResult.all().get();
        stringTopicDescriptionMap.forEach((k, v) -> {
            System.out.println(k + "\t" + v);
            System.out.println("=======================================");
            System.out.println(k);
            boolean internal = v.isInternal();
            String name = v.name();
            List<TopicPartitionInfo> partitions = v.partitions();
            String partitionStr = Arrays.toString(partitions.toArray());
            System.out.println("内部的？" + internal);
            System.out.println("topic name = " + name);
            System.out.println("分区：" + partitionStr);
            partitions.forEach(partition -> {
                System.out.println(partition);
            });
        });

    }

    // 查看集群信息
    @Test
    public void testDescribeCluster() throws ExecutionException,
            InterruptedException {
        DescribeClusterResult describeClusterResult = client.describeCluster();
        KafkaFuture<String> stringKafkaFuture = describeClusterResult.clusterId();
        String s = stringKafkaFuture.get();
        System.out.println("cluster name = " + s);
        KafkaFuture<Node> controller = describeClusterResult.controller();
        Node node = controller.get();
        System.out.println("集群控制器：" + node);
        Collection<Node> nodes = describeClusterResult.nodes().get();
        nodes.forEach(node1 -> {
            System.out.println(node1);
        });
    }

    // 查看配置信息
    @Test
    public void testDescribeConfigs() throws ExecutionException,
            InterruptedException, TimeoutException {
        ConfigResource configResource = new ConfigResource(ConfigResource.Type.BROKER, "0");
        DescribeConfigsResult describeConfigsResult
                = client.describeConfigs(Collections.singleton(configResource));
        Map<ConfigResource, Config> configMap = describeConfigsResult.all().get(15, TimeUnit.SECONDS);
        configMap.forEach(new BiConsumer<ConfigResource, Config>() {
            @Override
            public void accept(ConfigResource configResource, Config config) {
                ConfigResource.Type type = configResource.type();
                String name = configResource.name();
                System.out.println("资源名称：" + name);
                Collection<ConfigEntry> entries = config.entries();
                entries.forEach(new Consumer<ConfigEntry>() {
                    @Override
                    public void accept(ConfigEntry configEntry) {
                        boolean aDefault = configEntry.isDefault();
                        boolean readOnly = configEntry.isReadOnly();
                        boolean sensitive = configEntry.isSensitive();
                        String name1 = configEntry.name();
                        String value = configEntry.value();
                        System.out.println("是否默认：" + aDefault + "\t是否只读？" +
                                readOnly + "\t是否敏感？" + sensitive
                                + "\t" + name1 + " --> " + value);
                    }
                });
                ConfigEntry retries = config.get("retries");
                if (retries != null) {
                    System.out.println(retries.name() + " -->" + retries.value());
                } else {
                    System.out.println("没有这个属性");
                }
            }
        });
    }
    // 修改配置信息
    @Test
    public void testAlterConfig() throws ExecutionException, InterruptedException {
        // 这里设置后，原来资源中不冲突的属性也会丢失，直接按照这里的配置设置
        Map<ConfigResource, Config> configMap = new HashMap<>();
        ConfigResource resource = new ConfigResource(ConfigResource.Type.TOPIC,
                "admin_tp_01");
        Config config = new Config(Collections.singleton(new ConfigEntry("segment.bytes", "1048576000")));
        configMap.put(resource, config);
        AlterConfigsResult alterConfigsResult = client.alterConfigs(configMap);
        Void aVoid = alterConfigsResult.all().get();
    }
}

	
   